package mr

import (
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"
)
import "net"
import "os"
import "net/rpc"
import "net/http"

// task
type Task struct {
	//TaskType TaskType // 任务类型判断到底是map还是reduce
	//TaskId   int      // 任务的id
	Filename string
	//描述是否完成
	State    int
	Idmap    int
	Idreduce int
	// 传入的reducer的数量，用于hash
	// 输入文件

}
type Coordinator struct {

	// Your definitions here.
	State          int32 //状态表示是map还是reduce  0start 1map 2reduce
	NMapTask       int
	NReduceTask    int
	Maptask        chan Task
	Reducetask     chan Task
	MapTaskTime    sync.Map
	ReduceTaskTime sync.Map
	files          []string
	Mu             sync.Mutex //只能有一个worker访问

}
type TimeStamp struct {
	Time int64
	Fin  bool
}

func lenSyncMap(m *sync.Map) int {
	var i int
	m.Range(func(k, v interface{}) bool {
		i++
		return true
	})
	return i
}
func lenTaskFin(m *sync.Map) int { //获取同步映射中已完成任务的数量的
	var i int
	m.Range(func(k, v interface{}) bool {
		if v.(TimeStamp).Fin {
			i++
		}
		return true
	})

	return i
}

// Your code here -- RPC handlers for the worker to call.

// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
func (c *Coordinator) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}
func (c *Coordinator) GetTask(args *TaskRequest, reply *TaskResponse) error {
	state := atomic.LoadInt32(&c.State) //使用 atomic.LoadInt32(&c.State)，原子性地获取 c.State 的当前值。
	if state == 0 {
		if len(c.Maptask) != 0 {
			maptask, ok := <-c.Maptask
			fmt.Print("test\n")
			if ok {
				reply.XTask = maptask
			}
			reply.CurNMapTask = len(c.Maptask)

		} else {
			reply.CurNMapTask = -1
		}
		reply.CurNReducetask = len(c.Reducetask)
	} else if state == 1 {
		if len(c.Reducetask) != 0 {
			fmt.Printf("reduce task")
			reducetask, rok := <-c.Reducetask
			if rok {
				reply.XTask = reducetask
			}
			reply.CurNMapTask = -1
			reply.CurNReducetask = len(c.Reducetask)
		} else {
			reply.CurNMapTask = -1
			reply.CurNReducetask = -1
		}
	}

	reply.NReduceTask = c.NReduceTask
	reply.NMapTask = c.NMapTask
	reply.State = c.State

	//mr-X-Y
	return nil
}
func (c *Coordinator) TaskFin(args *Task, reply *ExampleReply) error {

	time_now := time.Now().Unix() //获取控制节点的时间
	if lenTaskFin(&c.MapTaskTime) != c.NMapTask {
		start_time, _ := c.MapTaskTime.Load(args.Idmap)
		if time_now-start_time.(TimeStamp).Time > 10 {
			return nil
		}
		c.MapTaskTime.Store(args.Idmap, TimeStamp{time_now, true})
		if lenTaskFin(&c.MapTaskTime) == c.NMapTask {
			atomic.StoreInt32(&c.State, 1)
			for i := 0; i < c.NReduceTask; i++ {
				c.Reducetask <- Task{Idreduce: i}
				c.ReduceTaskTime.Store(i, TimeStamp{time_now, false})
			}
		}
		//	fmt.Println("len%v<%v\n", len(c.MapTaskFin), c.NMapTask)
		//c.MapTaskFin <- true
		//if len(c.MapTaskFin) == c.NMapTask {
		//	fmt.Println("len%v<%v\n", len(c.MapTaskFin), c.NMapTask)
		//	c.State = 1
		//}
	} else if lenTaskFin(&c.ReduceTaskTime) != c.NReduceTask {
		start_time, _ := c.ReduceTaskTime.Load(args.Idmap)
		if time_now-start_time.(TimeStamp).Time > 10 {
			return nil
		}
		c.ReduceTaskTime.Store(args.Idreduce, TimeStamp{time_now, true})
		if lenTaskFin(&c.ReduceTaskTime) == c.NReduceTask {
			atomic.StoreInt32(&c.State, 2)
		}
		//c.ReduceTaskFin <- true
		//fmt.Printf("%v,%v", len(c.ReduceTaskFin), c.NReduceTask)
		//if len(c.ReduceTaskFin) == c.NReduceTask {
		//	c.State = 2
		//	fmt.Print("------------------->fin")
		//}
	}

	return nil
}

// start a thread that listens for RPCs from worker.go
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
func (c *Coordinator) Done() bool {
	state := atomic.LoadInt32(&c.State)
	time_now := time.Now().Unix()

	if state == 0 {
		for i := 0; i < c.NMapTask; i++ {
			tmp, _ := c.MapTaskTime.Load(i)
			if !tmp.(TimeStamp).Fin && time_now-tmp.(TimeStamp).Time > 10 {
				fmt.Print("map time out")
				c.Maptask <- Task{Filename: c.files[i], Idmap: i}
				c.MapTaskTime.Store(i, TimeStamp{time_now, false})
			}
		}
	} else if state == 1 {
		for i := 0; i < c.NReduceTask; i++ {
			tmp, _ := c.ReduceTaskTime.Load(i)
			if !tmp.(TimeStamp).Fin && time_now-tmp.(TimeStamp).Time > 10 {
				fmt.Print("reduce time out")
				c.Reducetask <- Task{Idreduce: i}
				c.ReduceTaskTime.Store(i, TimeStamp{time_now, false})
			}
		}
	}
	ret := false
	if lenTaskFin(&c.ReduceTaskTime) == c.NReduceTask {
		ret = true
	}
	// Your code here.
	//time.Sleep(200 * time.Millisecond)
	//var ret bool
	//c.Mu.Lock()
	//if c.State == 2 {
	//	ret = true
	//} else {
	//	ret = false
	//}
	//c.Mu.Unlock()
	return ret
	//time.Sleep(200 * time.Millisecond)
}

// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{
		State:       0, //状态表示是map还是reduce  0start 1map 2reduce
		NMapTask:    len(files),
		NReduceTask: nReduce,
		Maptask:     make(chan Task, len(files)),
		Reducetask:  make(chan Task, nReduce),
		files:       files}
	time_now := time.Now().Unix()

	//通过循环把所有的task都传入Maptask这个信道里面
	for id, file := range files {
		c.Maptask <- Task{Filename: file, Idmap: id}
		c.MapTaskTime.Store(id, TimeStamp{time_now, false})
	}
	c.server()
	return &c
}
func (c *Coordinator) TImeTick() {
	state := atomic.LoadInt32(&c.State)
	time_now := time.Now().Unix()

	if state == 0 {
		for i := 0; i < c.NMapTask; i++ {
			tmp, _ := c.MapTaskTime.Load(i)
			if !tmp.(TimeStamp).Fin && time_now-tmp.(TimeStamp).Time > 10 {
				fmt.Print("map time out")
				c.Maptask <- Task{Filename: c.files[i], Idmap: i}
				c.MapTaskTime.Store(i, TimeStamp{time_now, false})
			}
		}
	} else if state == 1 {
		for i := 0; i < c.NReduceTask; i++ {
			tmp, _ := c.ReduceTaskTime.Load(i)
			if !tmp.(TimeStamp).Fin && time_now-tmp.(TimeStamp).Time > 10 {
				fmt.Print("reduce time out")
				c.Reducetask <- Task{Idreduce: i}
				c.ReduceTaskTime.Store(i, TimeStamp{time_now, false})
			}
		}
	}

}
